package com.aotain.project.apollo.bolt;

import java.util.HashMap;
import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.mushroom.Slave;
import com.aotain.apollo.IPDatabase;

/**
 * 
 *  @ClassName: UDIISourceFmtBolt 
 * @Description: TODO(这里用一句话描述这个类的作用) 
 * @author 
 * @date 
 *
 */
public class HurstFmtBoltTest implements IRichBolt {

	/** 
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	 */ 
	private static final long serialVersionUID = 7816556443307732593L;

	
	OutputCollector collector = null;
	HashMap<Long, IPDatabase> map = null;
	String dip = null;
	String dport = null;
	String sip = null;
	String sport = null;
	String dnpacket = null;
	String uppacket = null;
	String dnoct = null;
	String upoct = null;
	String accesstime = null;	
	String delaytime = null;
	String logserver = null;
	String zooserver = null;
	
	public HurstFmtBoltTest(){}

	public HurstFmtBoltTest(String filterIPs) {
		//this.filterIPs = filterIPs;
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declareStream("UDIIStreamId",new Fields("dip","dport","sip","sport","dnpacket","uppacket","dnoct","upoct","delaytime"));
	}   

	@Override
	public Map<String, Object> getComponentConfiguration() {
		return null;
	}


	@Override
	public void execute(Tuple tuple) {
		countInLocal(tuple); 
	}

	private void countInLocal(Tuple tuple) {
		
		String logString = tuple.getString(0);
		if(logString != null && !"".equals(logString)) {			
			String[] splits = logString.split("\\|",-1);
			if(splits.length>20) {					
				dip = splits[9];
				dport = splits[10];
				sip = splits[12];
				sport = splits[13];
				dnoct = splits[17];
				upoct = splits[18];
				dnpacket = splits[19];
				uppacket = splits[20];				
				delaytime = 0 + "";
				long currenttime = System.currentTimeMillis();
				if(accesstime != null && !"".equals(accesstime)) {
					delaytime = currenttime/1000 - Long.parseLong(accesstime.trim()) + "";
				}
				collector.emit("UDIIStreamId",new Values(dip,dport,sip,sport,dnpacket,uppacket,dnoct,upoct,delaytime));
				collector.ack(tuple);
			}
		}
	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		if(!logserver.isEmpty())
	  	{
		  	Slave s = new Slave(zooserver);
		  
		  	s.SendAlarmLog("ERROR", "UDIISourceFmtBolt Shutdown");
	  	}
	}

	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		this.collector = collector;
	
		//String dbJson = stormConf.get("dbjson").toString();
		logserver = stormConf.get("mushroomserver").toString();	
		zooserver = stormConf.get("zkConn").toString();
		//ipsearch = new IPSearch(dbJson);
		//_ipseg = new IDCIPSegment(dbJson);
	}
}
